package cn.gwm.flink.streaming.dwd.streamsource;

import cn.gwm.flink.streaming.sink.kafka.KafkaConsumerEnum;
import cn.gwm.flink.streaming.sink.kafka.KafkaConsumerFactory;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import lombok.Data;
import lombok.experimental.Accessors;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

/**
 * @ClassName CommonStreamSource
 * @Description
 * @Author LiangGuang
 * @Date 2023/05/24 10:09
 */
@Data
@Accessors(chain = true)
public class CommonStreamSource extends StreamSource {

    public KafkaConsumerEnum consumerEnum;

    public Long startTime;

    public CommonStreamSource(KafkaConsumerEnum consumerEnum) {
        this.consumerEnum = consumerEnum;
    }

    public CommonStreamSource(KafkaConsumerEnum consumerEnum, String uidSuffix) {
        this.consumerEnum = consumerEnum;
        this.uidSuffix = uidSuffix;
    }


    @Override
    public DataStream<JSONObject> source(StreamExecutionEnvironment env, String groupId, String topic) {
        SingleOutputStreamOperator<JSONObject> sourceStream = env.addSource(getKafkaConsumer(consumerEnum, groupId, topic, startTime))
                .map(JSONUtil::parseObj).uid("convertJson_" + uidSuffix)
                .returns(JSONObject.class);
        return sourceStream;
    }


    public FlinkKafkaConsumer<String> getKafkaConsumer(KafkaConsumerEnum consumerEnum, String groupId, String topic, Long startTime) {
        FlinkKafkaConsumer<String> kafkaConsumer = KafkaConsumerFactory.getKafkaConsumer(consumerEnum, groupId, topic);
        if (KafkaConsumerEnum.TIMESTAMP.equals(consumerEnum)) {
            kafkaConsumer.setStartFromTimestamp(startTime == null ? System.currentTimeMillis() : startTime);
        }
        return kafkaConsumer;
    }


}
